package com.gugee.ins.data.analysis.listener;

import com.gugee.ins.data.analysis.config.InsAnalysisConfig;
import com.gugee.ins.data.analysis.service.*;
import com.gugee.ins.data.common.consumer.AbstractGugeeConsumerHandler;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;


/**
 * InsAnalysisAvatar Listener
 *
 * @author Huangxin
 */
@Component
@Slf4j
public class InsAnalysisAccountListener {

    /**
     * InsBloggerFollowingConfig
     */
    @Autowired
    private InsAnalysisConfig insAnalysisConfig;

    /**
     * insAnalysisAccountNormalService
     */
    @Autowired
    private InsAnalysisAccountNormalService insAnalysisAccountNormalService;

    /**
     * insAnalysisAccountUnusualService
     */
    @Autowired
    private InsAnalysisAccountUnusualService insAnalysisAccountUnusualService;

    /**
     * insAnalysisAccountCheckUnusualService
     */
    @Autowired
    private InsAnalysisAccountCheckUnusualService insAnalysisAccountCheckUnusualService;

    /**
     *
     */
    private InsAnalysisAccountNormalHandler analysisAccountNormalHandler = new InsAnalysisAccountNormalHandler();

    /**
     *
     */
    private InsAnalysisAccountUnusualHandler analysisAccountUnusualHandler = new InsAnalysisAccountUnusualHandler();

    /**
     *
     */
    private InsAnalysisAccountCheckUnusualHandler analysisAccountCheckUnusualHandler = new InsAnalysisAccountCheckUnusualHandler();

    /**
     * uid接口爬取body为null
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insAnalysisAccountUnusual.id}",
            topics = {"${kafka.topics.insBloggerAccountUnusual.online}"},
            containerFactory = "factoryInsAnalysisAccountUnusualAck")
    public void insAnalysisAccountUnusual(List<InsBloggerMessage> messages, Acknowledgment ack) {
        analysisAccountUnusualHandler.handle("InsAnalysisAccountUnusual",messages, insAnalysisConfig.getAcceptInsAnalysisAccountUnusualMsgTimeMinRange(),ack);
    }

    /**
     * uid接口爬取正常
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insAnalysisAccountNormal.id}",
            topics = {"${kafka.topics.insBloggerAccountNormal.online}"},
            containerFactory = "factoryInsAnalysisAccountNormalAck")
    public void insAnalysisAccountNormal(List<InsBloggerMessage> messages, Acknowledgment ack) {
        analysisAccountNormalHandler.handle("InsAnalysisAccountNormal",messages, insAnalysisConfig.getAcceptInsAnalysisAccountNormalMsgTimeMinRange(),ack);
    }

    /**
     * 爬取 uid 验证是否异常
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insAnalysisAccountCheckUnusual.id}",
            topics = {"${kafka.topics.insAnalysisAccountCheckUnusual.online}"},
            containerFactory = "factoryInsAnalysisAccountCheckUnusualAck")
    public void insAnalysisAccountCheckUnusual(List<InsBloggerMessage> messages, Acknowledgment ack) {
        analysisAccountCheckUnusualHandler.handle("InsAnalysisAccountCheckUnusual",messages, insAnalysisConfig.getAcceptInsAnalysisAccountCheckUnusualMsgTimeMinRange(),ack);
    }

    class InsAnalysisAccountNormalHandler extends AbstractGugeeConsumerHandler<InsBloggerMessage>{
        @Override
        public void dealMsg(List<InsBloggerMessage> messages) {
            insAnalysisAccountNormalService.handleAnalysisAccountNormal(messages);
        }
    }

    class InsAnalysisAccountUnusualHandler extends AbstractGugeeConsumerHandler<InsBloggerMessage>{
        @Override
        public void dealMsg(List<InsBloggerMessage> messages) {
            insAnalysisAccountUnusualService.handleAnalysisAccountUnusual(messages);
        }
    }

    class InsAnalysisAccountCheckUnusualHandler extends AbstractGugeeConsumerHandler<InsBloggerMessage>{
        @Override
        public void dealMsg(List<InsBloggerMessage> messages) {
            insAnalysisAccountCheckUnusualService.handleAnalysisAccountCheckUnusual(messages);
        }
    }
}
